package com.yh.csx.bsf.ods.fullload;

import lombok.Data;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.stereotype.Component;

import com.yh.csx.bsf.ods.utils.Constant;

/***
 * 任务管理
 */
@Data
@Component
public class TaskManager {

	private volatile static Map<String, Task> tasks = new HashMap<String, Task>();
	public static Map<String, CanalFullLoadConsumer> consumers = new ConcurrentHashMap<String, CanalFullLoadConsumer>();
	public synchronized void putSubTask(SubTask subTask) {
		String subTaskkey = subTask.getDb() + "_" + subTask.getTable();

		if (tasks.containsKey(subTask.targetTable)) {
			Task task = tasks.get(subTask.targetTable);
			subTask.setParent(task);
			if (!task.getSubTasks().containsKey(subTaskkey)) {
				task.getSubTasks().put(subTaskkey, subTask);
				subTask.setProcess(-1L);
				consumers.remove(subTaskkey);
			} else {
				SubTask tem = task.getSubTasks().get(subTaskkey);
				if (tem.process >= 1) {
					tem.setProcess(-1);	
					consumers.remove(subTaskkey);
				}
			}
		} else {
			Task task = new Task();
			task.getFinishedCount().set(0);
			subTask.setParent(task);
			subTask.setProcess(-1L);
			task.getSubTasks().put(subTaskkey, subTask);
			task.setTargetTable(subTask.targetTable);
			tasks.put(subTask.targetTable, task);
			consumers.remove(subTaskkey);
			
		}
	}

	/**
	 * 清理任务
	 */
	public void clear() {
		tasks.clear();
	}

	public Map<String, Task> getTasks() {
		return tasks;
	}

	public long getTaskCount() {
		return tasks.size();
	}

	public long getRunningCount() {
		return tasks.values().stream().filter(c -> c.getProgress() < 1 && c.getProgress() !=-1).count();
	}

	public long getDoneCount() {
		return tasks.values().stream().filter(c -> c.getProgress() >= Constant.PROGRESS_SUCCESS_CODE).count();
	}

	public long getErrorCount() {
		return tasks.values().stream().filter(c -> c.getProgress() == Constant.PROGRESS_ERROR_CODE).count();
	}

	public long getWaitingCount() {
		return tasks.values().stream().filter(c -> c.isWaited() == true).count();
	}

	public synchronized List<TaskManager.Task> setRunning(long num) {
		List<TaskManager.Task> newTasks = new ArrayList<TaskManager.Task>();
		tasks.values().stream().filter(c -> c.isStarted() == false).limit(num).forEach(c -> {
			newTasks.add(c);
		});
		return newTasks;
	}

	/**
	 * 同步任务
	 */
	@Data
	public static class Task {
		/**
		 * {db}.{table}
		 */
		private String key;
		private String targetDatabase;
		private String targetTable;
		private AtomicInteger taskCount = new AtomicInteger(0);
		private AtomicInteger finishedCount = new AtomicInteger(0);
		private Map<String, SubTask> subTasks = new ConcurrentHashMap<String, SubTask>();
	
		public Date getStartTime() {
			Date time=new Date();
			for (SubTask subTask : subTasks.values()) {
				if(subTask.getStartTime()!=null) {
					if(subTask.getStartTime().before(time)) {
						time=subTask.getStartTime();
					}
				}
			}
			return time;
		}
		public Date getEndTime() {
			Date time=new Date();
			Calendar c= Calendar.getInstance();
			c.set(2000, 1, 1);
			time=c.getTime();
			for (SubTask subTask : subTasks.values()) {
				if(subTask.getEndTime()!=null) {
					if(subTask.getEndTime().after(time)) {
						time=subTask.getEndTime();
					}
				}
			}
			return time;
		}
		/**
		 * 获取用时
		 * */
		public long getUsedTime() {
			return (Optional.ofNullable(this.getEndTime()).orElse(new Date()).getTime()-Optional.ofNullable(this.getStartTime()).orElse(new Date()).getTime())/1000;
		}
		public double getProgress() {
			double progress = 0;
			boolean started=false;
			for (SubTask subTask : subTasks.values()) {
				if (subTask.getProcess() == 999) {
					return 999;
				}
				if(subTask.getProcess()>=0) {
					started=true;
				}
				progress += subTask.getProcess();
			}
			if(started) {
				return progress / subTasks.size();
			}
			return -1;
		}

		public synchronized boolean isStarted() {
			for (SubTask subTask : subTasks.values()) {
				if (subTask.getProcess() >= 0)
					return true;
			}
			return false;
		}

		public synchronized boolean isSucceed() {
			return subTasks.values().stream().filter(c -> c.getProcess() >= 1 && c.getProcess() < 999)
					.count() == subTasks.values().size() ? true : false;
		}

		public boolean isWaited() {
			for (SubTask subTask : subTasks.values()) {
				if (subTask.getProcess() >= 0)
					return false;
			}
			return true;
		}
	}

	/**
	 * 子任务
	 */
	@Data
	public static class SubTask {
		private String db;
		private String table;
		private String targetTable;
		private volatile double process = -1L;
		private Date startTime;
		private Date endTime;
		private Task parent;

		public void setFail() {
			this.setProcess(999);
		}

		public void setSuccess() {
			this.setProcess(1);
		}

		public  void setProcess(double process) {
			if(process==0) {
				startTime=new Date();
			}
			if(process>=1&&process<999)
			{
				endTime=new Date();
			}
			this.process = process;
		}

	}
}
